package org.zjt.spark.dstream

import org.apache.spark.SparkConf
import org.apache.spark.streaming._

/**
  * DESC Window 时间窗口计算
  *
  *   reduceByWindow( handler,windowTime , actionTime)      执行窗口时间内的RDD
  *     每actionTime秒，执行一次windowTime接受时间内的任务。
  *
  *
  *   StreamingContext(conf, rddTime )                      切分RDD: windowTime/ rddTime
  *     将windowTime时间的数据除以rddTime切分为RDD数目。
  *
  *
  *
  *   PS: actionTime、windowTime是rddTime倍数，windowTime是actionTime的倍数
  *
  * @author
  * @create 2017-05-15 下午6:37
  **/
object DstreamWindowTest extends App {

  val conf = new SparkConf().setMaster("local[2]").setAppName("NetworkWordCount").set("spark.driver.allowMultipleContexts","true")
  val ssc = new StreamingContext(conf, Seconds(1))

  ssc.checkpoint("./checkpoint")
  // TODO: 1、以最近30秒作为一个输入窗口，每10秒执行一次。
  val resource = ssc.socketTextStream("localhost", 9002)

  val wordCounts = resource.window(Seconds(30),Seconds(10))
    .flatMap(_.split(" ")).map((_, 1))

    // TODO:  transform 对DStream中的每个RDD处理，结果RDD封装为DStream
    .transform(rdd =>rdd.map(x => (x._1.toUpperCase(),x._2))).reduceByKey(_+_)


  // TODO: 以最近30秒作为一个时间窗口（输入窗口），每10秒执行一次。将最近30/10次的结果叠加。（既是最近三次的提交任务数目）
  resource.countByWindow(Seconds(30),Seconds(10)).print()

  // TODO: 以最近30秒作为一个时间窗口（输入窗口），每10秒执行一次。将最近30/10次的结果叠加。（既是最近三次的提交任务数目）
  resource.reduceByWindow(_+_,Seconds(30),Seconds(10)).print()

  // TODO:  reduceByKeyAndWindow(reduce正向，反向reduce函数，窗口长度，窗口执行时间）
  resource.flatMap(_.split(" ")).map((_, 1)).reduceByKeyAndWindow(_+_, _ - _,Seconds(30),Seconds(10)).print()

  wordCounts.print()
  ssc.start()
  ssc.awaitTermination()
  ssc.stop()
}
